package com.rem.flink.flink8State;

import com.rem.flink.flink2Source.ClickSource;
import com.rem.flink.flink2Source.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * flink中数据源源不断到来和处理，由一个任务来维护并且并且计算输出结果的所有*数据*  就叫做这个任务的状态
 * 算子任务分为：
 * *     有状态算子：除当前状态外，还需要依赖其他数据得到结果，其他数据就是状态，最常见的就是之前到达的数据 例如sum、聚合函数、窗口算子
 * *     无状态算子：计算时不依赖其他数据 例如 map、fliter、flatMap
 * 状态的管理：
 * *     状态存在内存中 内存中存在访问权限，容错问题等 flink框架都解决了
 * 状态的分类：
 * *     原始状态：是自定义的，相当于就是开辟了一块内存，需要我们自己管理，实现状态的序列化和故障恢复
 * *     托管状态：由 Flink 统一管理的，状态的存储访问、故障恢复和重组等一系列问题都由 Flink 实现，我们只要调接口就可以
 * *托管状态分类：
 * **       算子状态Operator State：状态作用范围为当前算子任务实例，只对当前并行子任务实例有效，算子状态可以用在所有算子上，
 * **                              实际上和使用本地变量没什么区别，所以使用上还需要进一步实现CheckpointedFunction接口
 * **       按键分区状态Keyed State：根据输入的key来维护和访问的，所以只能定义在按键分区流(KeyedStream)中，也就是在keyBy之后才可以使用，
 * **                              通过富函数(Rich Function)定义的算子都可以使用keyed State.
 * **无论Keyed State 还是 Operator State，它们都是在本地实例上维护的，也就是说每个并行子任务维护着对应的状态，算子的子任务之间状态不共享。
 * **   算子状态：
 * ***          列表状态（ListState）
 * ***          联合状态（UnionListState）
 * ***          广播状态（BroadcastState）
 * **   按键分区状态：
 * ***          值状态（ValueState）
 * ***          列表状态（ListState）
 * ***          映射状态（MapState）
 * ***          归约状态（ReducingState）
 * ***          聚合状态（AggregatingState）
 * <p>
 * 状态生存时间TTKL：状态随着时间的增长，如果不加以x限制，最终会导致内存存储空间耗尽。所以给状态配置一个'生存时间' 访状态在内存中超出这个时间就会清除状态
 *
 *
 * * 状态持久化
 *      检查点checkpoint (默认禁用)                     用作故障恢复
 *      保存点savepoint  (是自定义的镜像保存 手动触发)     用作手动备份或恢复
 * * 状态后端
 *      哈希表状态后端 HashMapStateBackend   内存中
 *      内嵌RocksDB状态后端                  hdfs
 *
 * 分界线barier 使用分布式快照算法 (广播到下游)，不需要等所有算子任务都处理完成，省去等待时间
 *
 * @author Rem
 * @date 2022-10-24
 */

public class StateTests {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getTimestamp()));


        stream.keyBy(Event::getUser)
                .flatMap(new MyFlatMap())
                .print();

        env.execute();

    }

    /**
     * 按键分区状态Keyed State 测试
     */
    private static class MyFlatMap extends RichFlatMapFunction<Event, String> {
        ValueState<Event> valueState;
        ListState<Event> listState;
        MapState<String, Long> mapState;
        ReducingState<Event> reducingState;
        AggregatingState<Event, String> aggregatingState;

        @Override
        public void open(Configuration parameters) {
            RuntimeContext runtimeContext = super.getRuntimeContext();
            ValueStateDescriptor<Event> eventValueStateDescriptor = new ValueStateDescriptor<>("my-state", Event.class);
            valueState = runtimeContext.getState(eventValueStateDescriptor);

            /**
             * 配置状态ttl
             * 设置状态生存时间
             */
            StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(1))
                    /**
                     * 指定状态更新类型 型指定了什么时候更新状态失效时间，
                     * Disabled ：禁用ttl 永不过期
                     * OnCreateAndWrite：在创建和写的时候更新
                     * OnReadAndWrite：读写都会更新
                     */
                    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                    /**
                     * 设置状态的可见效
                     * 因为清除操作并不是实时的，所以当状态过期之后还有可能基于存在，这时如果对它进行访问，能否正常读取到就是一个问题了
                     * ReturnExpiredIfNotCleanedUp：未清理返回过期用户值
                     * NeverReturnExpired：永远不返回过期用户值
                     */
                    .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
                    .build();
            eventValueStateDescriptor.enableTimeToLive(ttlConfig);


            listState = runtimeContext.getListState(new ListStateDescriptor<>("my-list", Event.class));
            mapState = runtimeContext.getMapState(new MapStateDescriptor<>("my-map", String.class, Long.class));
            reducingState = runtimeContext.getReducingState(new ReducingStateDescriptor<>("my-reduce", (event1, event2) -> new Event(event1.getUser(), event1.getUrl(), event2.getTimestamp()), Event.class));
            aggregatingState = runtimeContext.getAggregatingState(new AggregatingStateDescriptor<>("my-agg", new AggregateFunction<>() {
                @Override
                public Long createAccumulator() {
                    return 0L;
                }

                @Override
                public Long add(Event value, Long accumulator) {
                    return accumulator + 1;
                }

                @Override
                public String getResult(Long accumulator) {
                    return "count:" + accumulator;
                }

                @Override
                public Long merge(Long a, Long b) {
                    return a + b;
                }
            }, Long.class));
        }


        // 增加一个本地变量进行对比
        Long count = 0L;

        @Override
        public void flatMap(Event event, Collector<String> out) throws Exception {
            System.out.println("访问 valueState：" + valueState.value());
            valueState.update(event);
            System.out.println("更新 valueState：" + valueState.value());

            System.out.println("访问 listState：" + listState.get());
            listState.add(event);
            System.out.println("更新 listState：" + listState.get());

            System.out.println("访问 mapState：" + mapState.get(event.getUser()));
            mapState.put(event.getUser(), mapState.get(event.getUser()) == null ? 1 : mapState.get(event.getUser()) + 1);
            System.out.println("更新 mapState：" + mapState.get(event.getUser()));

            System.out.println("访问 reducingState：" + reducingState.get());
            reducingState.add(event);
            System.out.println("更新 reducingState：" + reducingState.get());

            System.out.println("访问 aggregatingState：" + aggregatingState.get());
            aggregatingState.add(event);
            System.out.println("更新 aggregatingState：" + aggregatingState.get());

            count++;
            System.out.println("count: " + count);

        }
    }


}
